import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Set;

public class MyProcessor extends BaseRegionObserver {

    //重写postDelete方法，监听到relation表有删除好友操作时，实现双向好友关系删除，一方删除好友，另一方也会被迫删除好友
    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        //获取Table对象
        HTableInterface relation = e.getEnvironment().getTable(TableName.valueOf("relation"));
        //解析删除对象delete
        Set<Map.Entry<byte[], List<Cell>>> entries = delete.getFamilyCellMap().entrySet();
        for (Map.Entry<byte[], List<Cell>> entry : entries) {
            List<Cell> cells = entry.getValue();
            for (Cell cell : cells) {
                //获取rowkey
                byte[] rowKey = CellUtil.cloneRow(cell);
                //获取列名
                byte[] qualifier = CellUtil.cloneQualifier(cell);
                //准备get对象，用于判断即将要删除的目标数据是否存在
                Get get = new Get(qualifier);
                get.addColumn(Bytes.toBytes("friends"), rowKey);
                boolean flag = relation.exists(get);
                if (flag){
                    //如果目标数据存在，创建一个delete对象进行删除
                    Delete delete1 = new Delete(qualifier);
                    delete1.addColumn(Bytes.toBytes("friends"), rowKey);
                    relation.delete(delete1);
                }
            }
        }
        relation.close();
    }
}
